package com.lemon.core.mqapi.producer;

import com.lemon.core.common.lang.constant.Constants;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * @author lemon
 * @ClassName RabbitMQImpl
 * @desc
 * @Date 2019/3/8 19:14
 * @Version 1.0
 **/
public abstract class RabbitMQImpl {

    private ConnectionFactory factory = new ConnectionFactory();

    public RabbitMQImpl(String host, Integer port, String username, String password, String virtualHost) {
        factory.setHost(host);
        factory.setPort(port);
        factory.setUsername(username);
        factory.setPassword(password);
        factory.setVirtualHost(virtualHost);
    }

    /**
     *
     * @param queue 队列名称
     * @param data  消息内容
     */
    public void sendMessage(String queue, String data) {
        try {
            // 创建一个新的连接
            Connection connection = factory.newConnection();
            // 创建一个通道
            Channel channel = connection.createChannel();
            // 声明一个队列
            channel.queueDeclare(queue, true, false, false, null);

            AMQP.BasicProperties props = new AMQP.BasicProperties().builder().
                    contentType("text/plain").contentEncoding("UTF-8").deliveryMode(2).priority(0).build();

            channel.basicPublish("", queue, false, props, data.getBytes(Constants.UTF_8));

            channel.close();
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("发送消息失败，IOException，异常信息：" + e.getMessage());
        } catch (TimeoutException e) {
            e.printStackTrace();
            throw new RuntimeException("创建连接失败，TimeoutException，异常信息：" + e.getMessage());
        }
    }

    /**
     * 发送消息--延时队列
     * @param ttlQueue  死信队列
     * @param workQueue 工作队列
     * @param data      消息内容
     * @param ttlTime   延时时间(ms)
     */
    public void sendTTLMessage(String ttlQueue, String workQueue, String data, Long ttlTime) {
        try {
            // 创建一个新的连接
            Connection connection = factory.newConnection();
            // 创建一个通道
            Channel channel = connection.createChannel();
            // 配置
            Map<String, Object> args = new HashMap<>();
            args.put("x-dead-letter-exchange", "");
            args.put("x-dead-letter-routing-key", workQueue);

            channel.queueDeclare(workQueue, true, false, false, null);
            channel.queueDeclare(ttlQueue, true, false, false, args);
            // 发送消息
            channel.basicPublish("", ttlQueue,
                    new AMQP.BasicProperties.Builder().expiration(String.valueOf(ttlTime)).build(), data.getBytes(Constants.UTF_8));

            channel.close();
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("延迟发送消息失败，IOException，异常信息：" + e.getMessage());
        } catch (TimeoutException e) {
            e.printStackTrace();
            throw new RuntimeException("延迟创建连接失败，TimeoutException，异常信息：" + e.getMessage());
        }
    }
}
